CTA 5.0 - Optimization 2.0

import warnings

import time

import pandas as pd
import numpy as np

from cvx.simulator import Builder
from cvx.simulator import interpolate


from tinycta.linalg import solve, inv_a_norm
from tinycta.signal import returns_adjust, osc, shrink2id

warnings.simplefilter(action="ignore", category=FutureWarning)

# Load prices
prices = pd.read_csv("data/Prices_hashed.csv", index_col=0, parse_dates=True)

# interpolate the prices
prices = prices.apply(interpolate)

from ipywidgets import Label, HBox, VBox, IntSlider, FloatSlider

fast = IntSlider(min=4, max=192, step=4, value=32)
slow = IntSlider(min=4, max=192, step=4, value=96)
vola = IntSlider(min=4, max=192, step=4, value=32)
winsor = FloatSlider(min=1.0, max=6.0, step=0.1, value=4.2)
corr = IntSlider(min=50, max=500, step=10, value=200)
shrinkage = FloatSlider(min=0.0, max=1.0, step=0.05, value=0.5)
left_box = VBox(
    [
        Label("Fast Moving Average"),
        Label("Slow Moving Average"),
        Label("Volatility"),
        Label("Winsorizing"),
        Label("Correlation"),
        Label("Shrinkage"),
    ]
)

right_box = VBox([fast, slow, vola, winsor, corr, shrinkage])
HBox([left_box, right_box])

T = time.time()
correlation = corr.value

returns_adj = prices.apply(returns_adjust, com=vola.value, clip=winsor.value)

# DCC by Engle
cor = returns_adj.ewm(com=correlation, min_periods=correlation).corr()

mu = np.tanh(returns_adj.cumsum().apply(osc)).values
vo = prices.pct_change().ewm(com=vola.value, min_periods=vola.value).std().values

builder = Builder(prices=prices, initial_aum=1e8)

for n, (t, state) in enumerate(builder):
    mask = state.mask
    matrix = shrink2id(cor.loc[t[-1]].values, lamb=shrinkage.value)[mask, :][:, mask]
    expected_mu = np.nan_to_num(mu[n][mask])
    expected_vo = np.nan_to_num(vo[n][mask])
    risk_position = solve(matrix, expected_mu) / inv_a_norm(expected_mu, matrix)
    builder.cashposition = 1e6 * risk_position / expected_vo
    builder.aum = state.aum

portfolio = builder.build()
print(time.time() - T)

/home/runner/work/cs/cs/.venv/lib/python3.12/site-packages/pandas/core/arraylike.py:399: RuntimeWarning: invalid value encountered in log
  result = getattr(ufunc, method)(*inputs, **kwargs)

27.04198670387268

Conclusions

• Dramatic relativ improvements observable despite using the same signals as in previous Experiment.

• Main difference here is to take advantage of cross-correlations in the risk measurement.

• Possible to add constraints on individual assets or groups of them.

• Possible to reflect trading costs in objective with regularization terms (Ridge, Lars, Elastic Nets, …)

portfolio.snapshot()